package cn.doitedu.rtdw.data_dashboard;

import com.alibaba.fastjson.JSON;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import redis.clients.jedis.Jedis;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/6
 * @Desc: 学大数据，到多易教育
 * <p>
 * 每 5分钟输出一次最近30分钟内的新用户数

 -- 测试数据：
{"user_id":3,"username":"windy","session_id":"s10","event_id":"app_launch","event_time":1680763371000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"index","keyword":"usb 移动固态","search_id":"sc01"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"event1","event_time":1680763372000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"index","res_cnt":276,"search_id":"sc01"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"page_load","event_time":1680763373000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"index","search_id":"sc01","item_seq":1,"item_att":"ad"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"page_load","event_time":1680763374000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","search_id":"sc01","item_seq":3,"item_att":"nt"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"event3","event_time":1680763375000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","keyword":"固态移动硬盘","search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s10","event_id":"event4","event_time":1680763376000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"page_load","event_time":1680763377000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"wake_up","event_time":1680763378000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"event5","event_time":1680763379000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"page_load","event_time":1680763381000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"page_load","event_time":1680763382000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page09","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":4,"username":"jacky","session_id":"s10","event_id":"page_load","event_time":1680763383000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763384000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"index","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763385000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763386000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763387000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"index","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763388000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763389000,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":5,"username":"brown","session_id":"s11","event_id":"page_load","event_time":1680763389100,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s11","event_id":"page_load","event_time":1680763389200,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
{"user_id":3,"username":"windy","session_id":"s11","event_id":"page_load","event_time":1680763389300,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}

 -- 触发数据：
{"user_id":3,"username":"windy","session_id":"s11","event_id":"event6","event_time":1680763529300,"lat":38.089969323508726,"lng":114.35731900345093,"release_channel":"华为应用市场","device_type":"mi8","properties":{"url":"page08","res_cnt":276,"search_id":"sc02"},"register_phone":"18061581848","user_status":1,"register_time":"2018-08-03 16:46:38","register_gender":1,"register_birthday":"2002-03-06","register_province":"上海","register_city":"上海","register_job":"程序员","register_source_type":1,"gps_province":"河北省","gps_city":"石家庄市","gps_region":"鹿泉区","page_type":"文章页","page_service":"内容服务"}
 
 *
 **/
public class Dash01_TrafficOverview4 {
    public static void main(String[] args) throws Exception {
        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);
        env.setStateBackend(new HashMapStateBackend());
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建kafka的dwd数据连接器映射表
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka (                               "
                        + "     user_id           BIGINT,                     "
                        + "     username          string,                     "
                        + "     session_id        string,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     lat               double,                     "
                        + "     lng               double,                     "
                        + "     release_channel   string,                     "
                        + "     device_type       string,                     "
                        + "     properties        map<string,string>,         "
                        + "     register_phone    STRING,                     "
                        + "     user_status       INT,                        "
                        + "     register_time     TIMESTAMP(3),               "
                        + "     register_gender   INT,                        "
                        + "     register_birthday DATE,                       "
                        + "     register_province STRING,                     "
                        + "     register_city        STRING,                  "
                        + "     register_job         STRING,                  "
                        + "     register_source_type INT,                     "
                        + "     gps_province STRING,                          "
                        + "     gps_city     STRING,                          "
                        + "     gps_region   STRING,                          "
                        + "     page_type    STRING,                          "
                        + "     page_service STRING,                          "
                        + "     rt AS to_timestamp_ltz(event_time,3),         "
                        + "     WATERMARK FOR rt AS rt - interval '0' second  "
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd-events-detail',                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           ");

        // 写一个统计sql，实现需求逻辑
        tenv.executeSql(
                "  WITH tmp AS (                                                                                     "
                        +"  SELECT                                                                                            "
                        +"    user_id,                                                                                        "
                        +"    properties['url'] as page_url,                                                                  "
                        +"    page_type,                                                                                      "
                        +"    rt                                                                                              "
                        +"  FROM dwd_kafka                                                                                    "
                        +"  WHERE event_id = 'page_load')                                                                     "
                        +"                                                                                                    "
                        +" SELECT                                                                                             "
                        +"     window_start,                                                                                  "
                        +" 	window_end,                                                                                       "
                        +" 	page_type,                                                                                        "
                        +" 	page_url,                                                                                         "
                        +" 	page_uv                                                                                           "
                        +" FROM (                                                                                             "
                        +" SELECT                                                                                             "
                        +"  window_start,                                                                                     "
                        +" 	window_end,                                                                                       "
                        +" 	page_type,                                                                                        "
                        +" 	page_url,                                                                                         "
                        +" 	page_uv,                                                                                          "
                        +" 	row_number() over(partition by window_start,window_end,page_type order by page_uv desc) as rn     "
                        +" FROM ( 																		                      "
                        +"  SELECT                                                                                            "
                        +"     window_start,                                                                                  "
                        +" 	window_end,                                                                                       "
                        +" 	page_type,                                                                                        "
                        +" 	page_url,                                                                                         "
                        +" 	count(distinct user_id) as page_uv                                                                "
                        +"  FROM TABLE(                                                                                       "
                        +"    TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '5' MINUTE)                                            "
                        +"  )                                                                                                 "
                        +"  GROUP BY                                                                                          "
                        +"     window_start,                                                                                  "
                        +" 	window_end,                                                                                       "
                        +" 	page_type,                                                                                        "
                        +" 	page_url                                                                                          "
                        +" ) o1                                                                                               "
                        +" ) o2                                                                                               "
                        +" WHERE rn <= 2                                                                                     "
                ).print();

        env.execute();

    }
}
